package com.generator.service;

import com.generator.config.ServiceConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;

@Slf4j
@Service
public class ZkService implements InitializingBean {

    public static final String DEFAULT_CHARSET = "utf-8";

    @Resource
    private ServiceConfig serviceConfig;

    private CuratorFramework client;

    @Override
    public void afterPropertiesSet() throws Exception {

        RetryPolicy retryPolicy = new ExponentialBackoffRetry(serviceConfig.getCoordinatorRetryInterval(),serviceConfig.getCoordinatorRetryTimes());
        client = CuratorFrameworkFactory.builder()
                .connectString(serviceConfig.getCoordinatorAddress())
                .connectionTimeoutMs(serviceConfig.getCoordinatorConnectionTimeout())
                .sessionTimeoutMs(serviceConfig.getCoordinatorSessionTimeout())
                .retryPolicy(retryPolicy)
                .build();

        client.start();

    }

    /**
     * 创建节点
     * @param path 节点路径
     * @param data 数据
     * @param createMode 节点类型
     * @return boolean
     */
    public boolean createPath(String path, String data, CreateMode createMode) {

        try {

            byte[] byteData = StringUtils.isBlank(data) ? new byte[0] : data.getBytes(DEFAULT_CHARSET);
            client.create().creatingParentsIfNeeded().withMode(createMode).forPath(path,byteData);
            return true;
        } catch (Exception e) {
            resolveExceptionOnCreate(path,e);
        }

        return false;
    }

    public boolean createPath(String path,CreateMode createMode) {

        try {
            client.create().creatingParentsIfNeeded().withMode(createMode).forPath(path);
            return true;
        } catch (Exception e) {
            resolveExceptionOnCreate(path,e);
        }

        return false;
    }

    /**
     * 获取节点数据
     * @param path 节点路径
     * @return string
     */
    public String getData(String path) {

        try {
            byte[] data = client.getData().forPath(path);
            if(data == null || data.length < 1) {
                return "";
            }

            return new String(data,DEFAULT_CHARSET);
        } catch (Exception e) {
            log.error(String.format("[zookeeper] Getting value from node<%s> occur failure", path),e);
        }

        return "";
    }

    public boolean setData(String path,String data) {

        try {
            byte[] byteData = StringUtils.isBlank(data) ? new byte[0] : data.getBytes(DEFAULT_CHARSET);
            client.setData().forPath(path,byteData);
            return true;
        } catch (Exception e) {
            log.error(String.format("[zookeeper] Setting value to node<%s> occur failure", path),e);
        }

        return false;
    }

    public boolean exists(String path) {

        try {
            Stat stat = client.checkExists().forPath(path);
            return stat != null;
        } catch (Exception e) {
            log.error(String.format("[zookeeper] CheckingExists for node<%s> occur failure", path),e);
        }

        return false;
    }

    public PathChildrenCache createPathChildrenCache(String path) {
        return new PathChildrenCache(client,path,false);
    }

    public NodeCache createNodeCache(String path) {
        return new NodeCache(client,path);
    }

    public CuratorFramework getClient() {
        return client;
    }

    private void resolveExceptionOnCreate(String path,Exception e) {
        if(e instanceof KeeperException) {
            KeeperException ke = (KeeperException) e;
            if(ke.code() == KeeperException.Code.NODEEXISTS) {
                log.error("[zookeeper] The node {} exists.",path);
            }
        } else {
            log.error(String.format("[zookeeper] create node<%s> occur failure", path),e);
        }
    }
}
